package com.gitee.source.practice

import java.sql.{Connection, DriverManager, PreparedStatement, ResultSet}

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.source.{RichParallelSourceFunction, SourceFunction}
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}

/*
  1. 创建自定义连接数据源RichParallelSourceFunction
  2. 重写open, close, cancel, run四个方法
  3. 在open 中创建连接和查询, 还记得mysql的连接方法吗? Connection, DriverManager, PreparedStatement
    TODO 为什么连接mysql是java.sql包下的,而不是com.mysql.jdbc下的呢?
  4. 在run中进行数据读取, 还记得如何读取数据吗? executeQuery: 查询的结果集,遍历next()
 */
object CustomerSourceMySQL {
  def main(args: Array[String]): Unit = {
    val senv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val data: DataStream[Product] = mysqlConsumer(senv)
    data.print()
    senv.execute()
  }

  def mysqlConsumer(senv: StreamExecutionEnvironment): DataStream[Product] = {
    // 设置并行度为1 ,读取数据不会重复
    val data: DataStream[Product] = senv.addSource(new MySqlSource()).setParallelism(1)
    data
  }
}

class MySqlSource extends RichParallelSourceFunction[Product]() {
  var isRunning: Boolean = true
  var conn: Connection = _
  var ps: PreparedStatement = _

  //关闭连接
  override def close(): Unit = {
    if (conn != null || !conn.isClosed) conn.close()
    if (ps != null) ps.close()
  }

  //开启mysql的连接
  override def open(parameters: Configuration): Unit = {
    conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/day21_db", "root", "root")
    val sql = "select * from product"
    ps = conn.prepareStatement(sql)
  }

  override def cancel(): Unit = {
    isRunning = false
  }

  override def run(ctx: SourceFunction.SourceContext[Product]): Unit = {
    while (isRunning){
      val result: ResultSet = ps.executeQuery()
      while (result.next()){
        val pid: Long = result.getLong("pid")
        val pname: String = result.getString("pname")
        val num: Int = result.getInt("num")
        val product = Product(pid,pname,num)
        ctx.collect(product)
      }
      Thread.sleep(3000)
    }
  }
}

case class Product(pid: Long, pname: String, num: Int)
